/*
 * 文件名：RealStreamUnitDaoImpl.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：qiuyc
 * 修改时间：2014年11月2日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.dao.impl;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.dbutils.BasicRowProcessor;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.dbutils.ResultSetHandler;
import org.apache.commons.dbutils.handlers.BeanHandler;
import org.apache.commons.dbutils.handlers.BeanListHandler;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.constant.Constant;
import com.ailk.bigdata.etl.common.server.database.JdbcUtil;
import com.ailk.bigdata.etl.common.server.database.beanprocessor.HumpMatcher;
import com.ailk.bigdata.etl.common.server.database.beanprocessor.MyBeanProcessor;
import com.ailk.bigdata.etl.common.server.model.DatabaseInfo;
import com.ailk.bigdata.etl.common.server.model.MetaTable;
import com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao;
import com.ailk.bigdata.etl.realstream.server.model.Column;
import com.ailk.bigdata.etl.realstream.server.model.KafkaInfo;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamFormat;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUnit;
import com.ailk.bigdata.etl.realstream.server.model.SocketInfo;
import com.ailk.bigdata.etl.realstream.server.model.Table;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description TODO
 * @author [qiuyc]
 * @version [版本号,2014年11月2日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RealStreamUnitDaoPGImpl implements RealStreamUnitDao
{
	private static final Logger logger = LoggerFactory.getLogger(RealStreamUnitDaoPGImpl.class);
	private static final String getAllUnit = "select * from td_realstream_unit";
	private static final String getAllPassivUnit = "SELECT * FROM td_realstream_unit WHERE stream_trans_mode=?";
	private static final String getAllSendKafkaUnit = "SELECT * FROM td_realstream_unit WHERE if_send_kafka=1";
	private static final String getAllCalcUnitCode = "SELECT ONLY_UNIT_CODE FROM TD_REALSTREAM_UNIT";
	private static final String getCalcUnitCodeById = "SELECT * FROM td_realstream_unit WHERE ONLY_UNIT_CODE=?";
	private static final String getMetaTableId = "select * from td_meta_table where meta_table_id=?";
	private static final String getColumnsByTableId = "select * from td_meta_column where meta_table_id=?";
	private static final String getSocketInfoById = "SELECT socket_ip,socket_port FROM td_stream_initiative a left join td_socket_info b on a.socket_id = b.socket_id WHERE a.ONLY_UNIT_CODE=?";
	private static final String getDataFormatPattern = "select * from td_realstream_format where data_format_id=?";
	private static final String getKakfaConfigById = "select * from TD_KAFKA_INFO where kafka_id=?";

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public List<RealStreamUnit> getAllUnit()
	{
		List<RealStreamUnit> allUnit = new ArrayList<RealStreamUnit>();
		Connection conn = null;
		try
		{
			conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<RealStreamUnit> beanListHandler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = (List<RealStreamUnit>) JdbcUtil.query(conn, false, beanListHandler, getAllUnit);
			for (RealStreamUnit unit : allUnit)
			{
				ResultSetHandler<MetaTable<DatabaseInfo>> beanHandler = new BeanHandler(MetaTable.class, new BasicRowProcessor(new MyBeanProcessor(
						new HumpMatcher())));
				MetaTable<DatabaseInfo> metaTable = JdbcUtil.query(conn, false, beanHandler, getMetaTableId, unit.getMetaTableId());
				Table table = new Table();
				table.setTableName(metaTable.getTableName());
				Object[] params = new Object[1];
				params[0] = metaTable.getMetaTableId();
				ResultSetHandler<RealStreamUnit> beanListHandler2 = new BeanListHandler(Column.class, new BasicRowProcessor(new MyBeanProcessor(
						new HumpMatcher())));
				List<Column> columns = (List<Column>) JdbcUtil.query(conn, false, beanListHandler2, getColumnsByTableId, params);
				table.setColumns(columns.toArray(new Column[] {}));
				unit.setTable(table);
			}
		}
		catch (Exception e)
		{
			logger.error("查询所有的实时处理流接口单元出错！！！", e);
		}
		finally
		{
			DbUtils.closeQuietly(conn);
		}
		return allUnit;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public List<RealStreamUnit> getAllStreamUnit()
	{
		List<RealStreamUnit> allUnit = new ArrayList<RealStreamUnit>();
		try
		{
			Connection conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<RealStreamUnit> beanListHandler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = (List<RealStreamUnit>) JdbcUtil.query(conn, true, beanListHandler, getAllUnit);
		}
		catch (Exception e)
		{
			logger.error("查询所有的实时处理流接口单元出错！！！", e);
		}
		return allUnit;
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao#getAllCalcUnitCode()
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public List<RealStreamUnit> getAllCalcUnitCode()
	{
		List<RealStreamUnit> allUnit = null;
		try
		{
			Connection conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<List<RealStreamUnit>> handler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = JdbcUtil.query(conn, true, handler, getAllCalcUnitCode);
		}
		catch (Exception e)
		{
			logger.error("查询所有的实时处理流接口单元出错！！！", e);
		}
		return allUnit;
	}

	/**
	 * 
	 * @see com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao#getUnitByUnitCode(long)
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public RealStreamUnit getUnitByUnitCode(long onlyUnitCode)
	{
		RealStreamUnit streamUnit = null;
		Connection conn = null;
		try
		{
			conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			BeanHandler<RealStreamUnit> handler = new BeanHandler<RealStreamUnit>(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			streamUnit = JdbcUtil.query(conn, false, handler, getCalcUnitCodeById, onlyUnitCode);

			BeanHandler<RealStreamFormat> formatPattern = new BeanHandler<RealStreamFormat>(RealStreamFormat.class, new BasicRowProcessor(
					new MyBeanProcessor(new HumpMatcher())));
			RealStreamFormat sendFormat = JdbcUtil.query(conn, false, formatPattern, getDataFormatPattern, streamUnit.getSendFormatId());
			if (null != sendFormat)
			{
				streamUnit.setSendFormat(sendFormat);
			}

			// 查询对应表元数据
			ResultSetHandler<MetaTable<DatabaseInfo>> beanHandler = new BeanHandler(MetaTable.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			MetaTable<DatabaseInfo> metaTable = JdbcUtil.query(conn, false, beanHandler, getMetaTableId, streamUnit.getMetaTableId());
			if (null != metaTable)
			{
				Table table = new Table();
				table.setTableName(metaTable.getTableName());
				Object[] params = new Object[1];
				params[0] = metaTable.getMetaTableId();
				ResultSetHandler<RealStreamUnit> beanListHandler2 = new BeanListHandler(Column.class, new BasicRowProcessor(new MyBeanProcessor(
						new HumpMatcher())));
				List<Column> columns = (List<Column>) JdbcUtil.query(conn, false, beanListHandler2, getColumnsByTableId, params);
				table.setColumns(columns.toArray(new Column[] {}));
				streamUnit.setTable(table);
			}
		}
		catch (Exception e)
		{
			logger.error("查询实时处理流接口单元出错！！！", e);
		}
		finally
		{
			DbUtils.closeQuietly(conn);
		}
		return streamUnit;
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public RealStreamUnit getUnitAllByUnitCode(long onlyUnitCode)
	{
		RealStreamUnit streamUnitAll = new RealStreamUnit();

		Connection conn = null;
		try
		{
			conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			BeanHandler<RealStreamUnit> handler = new BeanHandler<RealStreamUnit>(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			streamUnitAll = JdbcUtil.query(conn, false, handler, getCalcUnitCodeById, onlyUnitCode);
			BeanHandler<RealStreamFormat> formatPattern = new BeanHandler<RealStreamFormat>(RealStreamFormat.class, new BasicRowProcessor(
					new MyBeanProcessor(new HumpMatcher())));
			RealStreamFormat acceptFormat = JdbcUtil.query(conn, false, formatPattern, getDataFormatPattern, streamUnitAll.getAcceptFormatId());
			if (null != acceptFormat)
			{
				streamUnitAll.setAcceptFormat(acceptFormat);
			}
			RealStreamFormat sendFormat = JdbcUtil.query(conn, false, formatPattern, getDataFormatPattern, streamUnitAll.getSendFormatId());
			if (null != sendFormat)
			{
				streamUnitAll.setSendFormat(sendFormat);
			}
			BeanListHandler<SocketInfo> handlerSocket = new BeanListHandler<SocketInfo>(SocketInfo.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			List<SocketInfo> socketInfo = (List<SocketInfo>) JdbcUtil.query(conn, false, handlerSocket, getSocketInfoById, onlyUnitCode);
			if (CollectionUtils.isNotEmpty(socketInfo))
			{
				streamUnitAll.setSocketInfo(socketInfo);
			}

			// 查询对应表元数据
			ResultSetHandler<MetaTable<DatabaseInfo>> beanHandler = new BeanHandler(MetaTable.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			MetaTable<DatabaseInfo> metaTable = JdbcUtil.query(conn, false, beanHandler, getMetaTableId, streamUnitAll.getMetaTableId());
			if (null != metaTable)
			{
				Table table = new Table();
				table.setTableName(metaTable.getTableName());
				Object[] params = new Object[1];
				params[0] = metaTable.getMetaTableId();
				ResultSetHandler<RealStreamUnit> beanListHandler2 = new BeanListHandler(Column.class, new BasicRowProcessor(new MyBeanProcessor(
						new HumpMatcher())));
				List<Column> columns = (List<Column>) JdbcUtil.query(conn, false, beanListHandler2, getColumnsByTableId, params);
				List<String> columnContains = new ArrayList<String>();
				for (Column metaColumn : columns)
				{
					columnContains.add(metaColumn.getColumnName().toLowerCase());
				}
				table.setColumns(columns.toArray(new Column[] {}));
				streamUnitAll.setTable(table);
			}
		}
		catch (Exception e)
		{
			streamUnitAll = null;
			logger.error("查询实时处理流接口单元出错！！！  onlyUnitCode[" + onlyUnitCode + "]", e);
		}
		finally
		{
			DbUtils.closeQuietly(conn);
		}
		return streamUnitAll;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public List<RealStreamUnit> getAllPassivUnit()
	{
		List<RealStreamUnit> allUnit = new ArrayList<RealStreamUnit>();
		try
		{
			Connection conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<RealStreamUnit> beanListHandler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = (List<RealStreamUnit>) JdbcUtil.query(conn, true, beanListHandler, getAllPassivUnit, Constant.STREAM_TRANS_PASSIVE_MODE);
		}
		catch (Exception e)
		{
			logger.error("查询实时处理流接口单元出错！！！", e);
		}
		return allUnit;
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao#getAllInitiativeUnit()
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public List<RealStreamUnit> getAllInitiativeUnit()
	{
		List<RealStreamUnit> allUnit = new ArrayList<RealStreamUnit>();
		try
		{
			Connection conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<RealStreamUnit> beanListHandler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = (List<RealStreamUnit>) JdbcUtil.query(conn, false, beanListHandler, getAllPassivUnit, Constant.STREAM_TRANS_INITIATIVE_MODE);
			// 关联查询出socket info信息
			if (CollectionUtils.isNotEmpty(allUnit))
			{
				for (RealStreamUnit streamUnit : allUnit)
				{
					BeanListHandler<SocketInfo> handlerSocket = new BeanListHandler<SocketInfo>(SocketInfo.class, new BasicRowProcessor(
							new MyBeanProcessor(new HumpMatcher())));
					List<SocketInfo> socketInfo = (List<SocketInfo>) JdbcUtil.query(conn, true, handlerSocket, getSocketInfoById,
							streamUnit.getOnlyUnitCode());
					if (CollectionUtils.isNotEmpty(socketInfo))
					{
						streamUnit.setSocketInfo(socketInfo);
					}
				}
			}
		}
		catch (Exception e)
		{
			logger.error("查询实时处理流接口单元出错！！！", e);
		}
		return allUnit;
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao#getALLKafkaSendUnit()
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	@Override
	public List<RealStreamUnit> getALLKafkaSendUnit()
	{
		List<RealStreamUnit> allUnit = new ArrayList<RealStreamUnit>();
		Connection conn = null;
		try
		{
			conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			ResultSetHandler<RealStreamUnit> beanListHandler = new BeanListHandler(RealStreamUnit.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			allUnit = (List<RealStreamUnit>) JdbcUtil.query(conn, false, beanListHandler, getAllSendKafkaUnit);
			if (CollectionUtils.isNotEmpty(allUnit))
			{
				BeanHandler<KafkaInfo> formatPattern = new BeanHandler<KafkaInfo>(KafkaInfo.class, new BasicRowProcessor(new MyBeanProcessor(
						new HumpMatcher())));
				for (RealStreamUnit streamUnit : allUnit)
				{
					if (StringUtils.isEmpty(streamUnit.getKafkaId()))
					{
						logger.warn("配置错误，单元接口[{}][{}]配置错误，kafka_id 配置不能为空", streamUnit.getUnitName(), streamUnit.getOnlyUnitCode());
						continue;
					}
					KafkaInfo kafkaInfo = JdbcUtil.query(conn, false, formatPattern, getKakfaConfigById, streamUnit.getKafkaId());
					if (null != kafkaInfo)
					{
						logger.info("单元接口[{}]接口名称[{}]，对应kafka消息队列配置[{}]", streamUnit.getOnlyUnitCode(), streamUnit.getUnitName(),
								kafkaInfo.toString());
						streamUnit.setKafkaInfo(kafkaInfo);
					}
				}

			}
		}
		catch (Exception e)
		{
			logger.error("查询实时处理流接口单元出错！！！", e);
		}
		finally
		{
			DbUtils.closeQuietly(conn);
		}
		return allUnit;
	}

	public KafkaInfo getKafkaInfo(String kafkaId)
	{
		Connection conn = null;
		KafkaInfo kafkaInfo = null;
		try
		{
			conn = JdbcUtil.getConnectByPool(RealStreamTool.dbInfo);
			BeanHandler<KafkaInfo> formatPattern = new BeanHandler<KafkaInfo>(KafkaInfo.class, new BasicRowProcessor(new MyBeanProcessor(
					new HumpMatcher())));
			kafkaInfo = JdbcUtil.query(conn, false, formatPattern, getKakfaConfigById, kafkaId);
		}
		catch (Exception e)
		{
			logger.error("查询实时处理流接口单元出错！！！", e);
		}
		finally
		{
			DbUtils.closeQuietly(conn);
		}
		return kafkaInfo;
	}
}
